package main

import (
	"fmt"
	"github.com/Shopify/sarama"
	"github.com/pkg/errors"
	"strings"
	"sync"
	"time"
)

type KafkaProducer struct {
	producerObj sarama.AsyncProducer
	conf        *KafkaCfg
	successcb   FuncSuccessCallback
	errorcb     FuncErrorCallback
	bExit       bool
	wg          *sync.WaitGroup
}

func (this *KafkaProducer) Init(conf *KafkaCfg, successCB FuncSuccessCallback, errorCB FuncErrorCallback, compressionType sarama.CompressionCodec) (err error) {
	this.conf = conf
	this.successcb = successCB
	this.errorcb = errorCB

	this.bExit = false

	config := sarama.NewConfig()
	config.Producer.Flush.Messages = 10000
	config.Producer.Flush.Frequency = time.Second * 1
	config.Producer.Flush.MaxMessages = 50000
	config.Producer.RequiredAcks = sarama.WaitForLocal
	config.Producer.Partitioner = sarama.NewRandomPartitioner
	config.Version = sarama.V0_10_0_0
	config.Producer.Compression = compressionType //算法类型

	if successCB != nil && errorCB != nil {
		config.Producer.Return.Successes = true
		config.Producer.Return.Errors = true
		this.wg = &sync.WaitGroup{}
		this.wg.Add(1)
		go this.dataSuccessCB()
	}

	this.producerObj, err = sarama.NewAsyncProducer(strings.Split(conf.Producer.Brokers, ","), config)
	if err != nil {
		errString := fmt.Sprintf("生产者创建失败！err=%s", err.Error())
		return errors.New(errString)
	}
	return nil
}

func (this *KafkaProducer) Fini() {
	this.bExit = true
	this.producerObj.AsyncClose()
	this.wg.Wait()
}

func (this *KafkaProducer) dataSuccessCB() {
	defer this.wg.Done()
	last := time.Now()
	for {
		if this.bExit == true && time.Since(last) > time.Second*1 {
			//fmt.Println("数据生产回调线程退出！")
			return
		}

		if this.producerObj == nil {
			time.Sleep(time.Millisecond * 1)
			continue
		}

		select {
		case msg := <-this.producerObj.Successes():
			if msg != nil {
				if this.successcb != nil {
					this.successcb(msg)
				}
				last = time.Now()
			}
		case msg := <-this.producerObj.Errors():
			if msg != nil {
				//fmt.Println("数据生产失败！err=", msg.Err)
				if this.errorcb != nil {
					this.errorcb(msg)
				}
				last = time.Now()
			}
		case <-time.After(1 * time.Second):
			time.Sleep(time.Millisecond * 1)
		}
	}
}

func (this *KafkaProducer) ProducerAsync_Byte(data []byte, pUser interface{}) (err error) {
	return this.producerAsync(&sarama.ProducerMessage{Topic: this.conf.Producer.Topic, Key: nil, Value: sarama.ByteEncoder(data), Metadata: pUser})
}

func (this *KafkaProducer) ProducerAsync_String(data string, pUser interface{}) (err error) {
	return this.producerAsync(&sarama.ProducerMessage{Topic: this.conf.Producer.Topic, Key: nil, Value: sarama.StringEncoder(data), Metadata: pUser})
}

func (this *KafkaProducer) producerAsync(msg *sarama.ProducerMessage) (err error) {
	defer func() {
		errPanic := recover()
		if errPanic != nil {
			errString := fmt.Sprintf("生产错误：err=%s", errPanic)
			err = errors.New(errString)
		}
	}()

	msg.Timestamp = time.Now()
	this.producerObj.Input() <- msg

	return nil
}
